package mqtt

import (
	"context"
	"fmt"
	"strings"
	"time"

	mqtt "github.com/eclipse/paho.mqtt.golang"
	"github.com/project-flogo/core/data/coerce"
	"github.com/project-flogo/core/data/metadata"
	"github.com/project-flogo/core/support/log"
	"github.com/project-flogo/core/support/ssl"
	"github.com/project-flogo/core/trigger"
)

var triggerMd = trigger.NewMetadata(&Settings{}, &HandlerSettings{}, &Output{})

func init() {
	_ = trigger.Register(&Trigger{}, &Factory{})
}

// Trigger is simple MQTT trigger
type Trigger struct {
	handlers              map[string]*clientHandler
	defaultPublishHandler trigger.Handler
	onConnectHandler      trigger.Handler
	connectionLostHandler trigger.Handler
	settings              *Settings
	logger                log.Logger
	options               *mqtt.ClientOptions
	client                mqtt.Client
}
type clientHandler struct {
	//client mqtt.Client
	handler  trigger.Handler
	settings *HandlerSettings
}
type Factory struct {
}

func (*Factory) Metadata() *trigger.Metadata {
	return triggerMd
}

// New implements trigger.Factory.New
func (*Factory) New(config *trigger.Config) (trigger.Trigger, error) {
	s := &Settings{}

	err := metadata.MapToStruct(config.Settings, s, true)
	if err != nil {
		return nil, err
	}

	return &Trigger{settings: s}, nil
}

// Initialize implements trigger.Initializable.Initialize
func (t *Trigger) Initialize(ctx trigger.InitContext) error {
	t.logger = ctx.Logger()

	settings := t.settings
	options, err := t.initClientOption(settings)
	if err != nil {
		return err
	}
	t.options = options
	t.logger.Debugf("Client options: %v", options)

	options.SetDefaultPublishHandler(func(client mqtt.Client, msg mqtt.Message) {
		t.logger.Warnf("Recieved message on unhandled topic: %s", msg.Topic())
	})

	t.handlers = make(map[string]*clientHandler)

	for _, handler := range ctx.GetHandlers() {

		s := &HandlerSettings{}
		err := metadata.MapToStruct(handler.Settings(), s, true)
		if err != nil {
			return err
		}
		switch strings.ToUpper(s.Kind) {
		case "ONCONNECT":
			{
				t.onConnectHandler = handler
				break
			}
		case "ONCONNECTIONLOST":
			{
				t.connectionLostHandler = handler
				break
			}
		case "DEFAULTPUBLISH":
			{
				t.defaultPublishHandler = handler
			}
		default:
			{
				t.handlers[s.Topic] = &clientHandler{handler: handler, settings: s}
			}
		}
	}

	return nil
}

var DEFAULT_KEEP_ALIVE = 60 * time.Second

func (t *Trigger) initClientOption(settings *Settings) (*mqtt.ClientOptions, error) {

	opts := mqtt.NewClientOptions()
	if len(settings.Servers) == 0 {
		return nil, fmt.Errorf("invalid settings, servers are required")
	}
	for _, val := range settings.Servers {
		s, err := coerce.ToString(val)
		if err != nil {
			return nil, fmt.Errorf("invalid settings, server url should be string")
		}
		st := strings.TrimSpace(s)
		if st == "" {
			return nil, fmt.Errorf("invalid settings, server url should not be empty")
		}
		opts.AddBroker(st)
	}
	clientID := strings.TrimSpace(settings.ClientID)
	if clientID != "" {
		opts.SetClientID(clientID)
	}

	username := strings.TrimSpace(settings.Username)
	if username != "" {
		opts.SetUsername(username)
	}

	password := strings.TrimSpace(settings.Password)
	if password != "" {
		opts.SetPassword(password)
	}

	opts.SetCleanSession(settings.CleanSession)
	opts.SetOrderMatters(settings.Order)
	if settings.WillEnabled {
		willTopic := strings.TrimSpace(settings.WillTopic)
		if willTopic == "" {
			return nil, fmt.Errorf("invalid settings, lastWill enabled but will topic not set")
		}
		payload := make([]byte, len(settings.WillPayload))
		for idx, val := range settings.WillPayload {
			bi, err := coerceToInt(val)
			if err != nil {
				return nil, fmt.Errorf("invalid last will payload: %v", settings.WillPayload)
			}
			b := byte(bi)
			payload[idx] = b
		}
		opts.SetBinaryWill(willTopic, payload, byte(settings.WillQos), settings.WillRetained)
	}
	if settings.ProtocolVersion > 0 {
		opts.SetProtocolVersion(uint(settings.ProtocolVersion))
	}
	// if (settings.TLSConfig)

	keepAliveStr := strings.TrimSpace(settings.KeepAlive)
	if keepAliveStr != "" {
		keepAlive, err := time.ParseDuration(keepAliveStr)
		if err != nil {
			return nil, fmt.Errorf("invalid settings, invalid KeepAlive duration")
		}
		opts.SetKeepAlive(keepAlive)
	} else {
		opts.SetKeepAlive(DEFAULT_KEEP_ALIVE)
	}

	pingTimeoutStr := strings.TrimSpace(settings.PingTimeout)
	if pingTimeoutStr != "" {
		pingTimeout, err := time.ParseDuration(pingTimeoutStr)
		if err != nil {
			return nil, fmt.Errorf("invalid settings, invalid PingTimeout duration")
		}
		opts.SetPingTimeout(pingTimeout)
	}

	connectTimeoutStr := strings.TrimSpace(settings.ConnectTimeout)
	if connectTimeoutStr != "" {
		connectTimeout, err := time.ParseDuration(connectTimeoutStr)
		if err != nil {
			return nil, fmt.Errorf("invalid settings, invalid ConnectTimeout duration")
		}
		opts.SetConnectTimeout(connectTimeout)
	}

	maxConnectIntervalStr := strings.TrimSpace(settings.MaxReconnectInterval)
	if maxConnectIntervalStr != "" {
		maxConnectInterval, err := time.ParseDuration(settings.MaxReconnectInterval)
		if err != nil {
			return nil, fmt.Errorf("invalid settings, invalid MaxConnectInterval duration")
		}
		opts.SetMaxReconnectInterval(maxConnectInterval)
	}

	opts.SetAutoReconnect(settings.AutoReconnect)

	if settings.Store != ":memory:" && settings.Store != "" {
		opts.SetStore(mqtt.NewFileStore(settings.Store))
	}

	writeTimeoutStr := strings.TrimSpace(settings.WriteTimeout)
	if writeTimeoutStr != "" {
		writeTimeout, err := time.ParseDuration(settings.WriteTimeout)
		if err != nil {
			return nil, fmt.Errorf("invalid settings, invalid WriteTimeout duration")
		}
		opts.SetWriteTimeout(writeTimeout)
	}

	opts.SetResumeSubs(settings.ResumeSubs)

	if len(settings.HTTPHeaders) > 0 {
		headers := map[string][]string{}
		for k, v := range settings.HTTPHeaders {
			headers[k] = []string{v}
		}
		opts.SetHTTPHeaders(headers)
	}

	opts.SetDefaultPublishHandler(t.genDefaultPublishHandler())
	opts.SetOnConnectHandler(t.genOnConnectHandler())
	opts.SetConnectionLostHandler(t.genConnectionLostHandler())

	if len(settings.TLSConfig) != 0 {
		cfg := &ssl.Config{}
		err := cfg.FromMap(settings.TLSConfig)
		if err != nil {
			return nil, fmt.Errorf("invalid settings, invalid tls settings")
		}

		if _, set := settings.TLSConfig["skipVerify"]; !set {
			cfg.SkipVerify = true
		}
		if _, set := settings.TLSConfig["useSystemCert"]; !set {
			cfg.UseSystemCert = true
		}
		tlsConfig, err := ssl.NewClientTLSConfig(cfg)
		if err != nil {
			return nil, err
		}
		opts.SetTLSConfig(tlsConfig)
	}

	return opts, nil
}

func (t *Trigger) genDefaultPublishHandler() func(mqtt.Client, mqtt.Message) {
	return func(client mqtt.Client, message mqtt.Message) {
		_, err := runHandler(t.defaultPublishHandler, message)
		if err != nil {
			t.logger.Errorf("failed run defaultPublishHandler: %v", err)
		}
	}
}

func (t *Trigger) genOnConnectHandler() func(mqtt.Client) {
	return func(client mqtt.Client) {
		opr := client.OptionsReader()
		t.logger.Info("run OnConnect handler: %s", opr.ClientID())
		err := runOnConnectHandler(t.onConnectHandler, client)
		if err != nil {
			t.logger.Errorf("failed run OnConnect handler: %v", err)
		}
	}
}

func (t *Trigger) genConnectionLostHandler() func(mqtt.Client, error) {
	return func(client mqtt.Client, cerr error) {
		opr := client.OptionsReader()
		t.logger.Info("run ConnectionLost handler: %s, cause: %v", opr.ClientID(), cerr)
		err := runConnectionLostHandler(t.connectionLostHandler, client, cerr)
		if err != nil {
			t.logger.Errorf("failed run ConnectionLost handler: %s", err)
		}
	}
}

// Start implements trigger.Trigger.Start
func (t *Trigger) Start() error {

	client := mqtt.NewClient(t.options)

	if token := client.Connect(); token.Wait() && token.Error() != nil {
		return token.Error()
	}

	t.client = client

	for _, handler := range t.handlers {

		if token := client.Subscribe(handler.settings.Topic, byte(handler.settings.Qos), t.getHanlder(handler)); token.Wait() && token.Error() != nil {
			t.logger.Errorf("Error subscribing to topic %s: %s", handler.settings.Topic, token.Error())
			return token.Error()
		}

		t.logger.Debugf("Subscribed to topic: %s", handler.settings.Topic)
	}

	return nil
}

// Stop implements ext.Trigger.Stop
func (t *Trigger) Stop() error {

	//unsubscribe from topics
	for _, handler := range t.handlers {
		t.logger.Debug("Unsubscribing from topic: ", handler.settings.Topic)
		if token := t.client.Unsubscribe(handler.settings.Topic); token.Wait() && token.Error() != nil {
			t.logger.Errorf("Error unsubscribing from topic %s: %s", handler.settings.Topic, token.Error())
		}
	}

	t.client.Disconnect(250)

	return nil
}

func (t *Trigger) getHanlder(handler *clientHandler) func(mqtt.Client, mqtt.Message) {
	return func(client mqtt.Client, msg mqtt.Message) {
		topic := msg.Topic()
		qos := msg.Qos()

		t.logger.Debugf("Topic[%s] - Payload Recieved: %s", topic, msg.Payload())

		result, err := runHandler(handler.handler, msg)
		if err != nil {
			t.logger.Error("Error handling message: %v", err)
			return
		}
		msg.Ack()

		if handler.settings.ReplyTopic != "" {
			reply := &Reply{}
			err = reply.FromMap(result)
			if err != nil {
				t.logger.Error("Error handling message: %v", err)
				return
			}

			if reply.Data != nil {
				// dataJson, err := json.Marshal(reply.Data)
				// if err != nil {
				// 	return
				// }
				token := client.Publish(handler.settings.ReplyTopic, qos, false, reply.Data)
				sent := token.WaitTimeout(5000 * time.Millisecond)
				if !sent {
					t.logger.Errorf("Timeout occurred while trying to publish reply to topic '%s'", handler.settings.ReplyTopic)
					return
				}
			}
		}
	}
}

// RunHandler runs the handler and associated action
func runHandler(handler trigger.Handler, message mqtt.Message) (map[string]interface{}, error) {

	if handler != nil {
		out := &Output{}
		out.Duplicate = message.Duplicate()
		out.Qos = message.Qos()
		out.Retained = message.Retained()
		out.Topic = message.Topic()
		out.MessageID = message.MessageID()
		out.Payload = message.Payload()

		results, err := handler.Handle(context.Background(), out)
		if err != nil {
			return nil, err
		}

		return results, nil
	}
	return map[string]interface{}{}, nil
}

func runOnConnectHandler(handler trigger.Handler, client mqtt.Client) error {
	if handler != nil {
		var err error
		out := &Output{}
		out.IsConnected = client.IsConnected()
		out.IsConnectionOpen = client.IsConnectionOpen()
		opr := client.OptionsReader()
		out.ClientOptions = toClientOptions(opr)
		_, err = handler.Handle(context.Background(), out)
		return err
	}
	return nil
}

func runConnectionLostHandler(handler trigger.Handler, client mqtt.Client, cerr error) error {
	if handler != nil {
		out := &Output{}
		out.IsConnected = client.IsConnected()
		out.IsConnectionOpen = client.IsConnectionOpen()
		opr := client.OptionsReader()
		out.ClientOptions = toClientOptions(opr)
		out.ConnectionLostErr = cerr.Error()

		_, err := handler.Handle(context.Background(), out)

		return err
	}
	return nil
}

func toClientOptions(reader mqtt.ClientOptionsReader) clientOptions {
	options := clientOptions{}
	servers := reader.Servers()
	options.Servers = make([]string, len(servers))
	for idx, url := range servers {
		options.Servers[idx] = url.String()
	}

	options.ResumeSubs = reader.ResumeSubs()
	options.ClientID = reader.ClientID()
	options.Username = reader.Username()
	options.Password = reader.Password()
	options.CleanSession = reader.CleanSession()
	options.Order = reader.CleanSession()
	options.WillEnabled = reader.WillEnabled()
	options.WillTopic = reader.WillTopic()
	options.WillPayload = reader.WillPayload()
	options.WillQos = reader.WillQos()
	options.WillRetained = reader.WillRetained()
	options.ProtocolVersion = reader.ProtocolVersion()
	// options.TLSConfig = reader.TLSConfig()
	options.KeepAlive = reader.KeepAlive().String()
	options.PingTimeout = reader.PingTimeout().String()
	options.ConnectionTimeout = reader.ConnectTimeout().String()
	options.MaxReconnectInterval = reader.MaxReconnectInterval().String()
	// options.ConnectRetry = reader.Conn
	options.WriteTimeout = reader.WriteTimeout().String()
	options.MessageChannelDepth = reader.MessageChannelDepth()
	options.HTTPHeaders = reader.HTTPHeaders()

	return options
}
